package com.tinem.platform.service.message.observable;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.tinem.platform.module.pojo.enums.message.CommandEnum;
import com.tinem.platform.module.pojo.enums.message.Message;
import com.tinem.platform.module.pojo.enums.message.MqttMessageType;
import com.tinem.platform.module.pojo.enums.message.PushChannel;
import com.tinem.platform.module.starter.EmqxClient;
import com.tinem.platform.service.message.config.ClientConfiguration;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.ApplicationListener;
import org.springframework.core.annotation.Order;
import org.springframework.stereotype.Service;

import javax.annotation.Resource;
import java.util.Arrays;
import java.util.Map;
import java.util.stream.Collectors;

/**
 * @author fengzhihao
 * @version v1
 * @program: platform
 * @className WebSocketListener
 * @description TODO
 * @site
 * @company
 * @create 2021/12/1 17:11:03
 */
@Slf4j
@Service
@Order(100)
public class MqttListener implements ApplicationListener<MessageObservable> {

    @Value("${spring.profiles.active}")
    protected String active;

    @Resource
    ClientConfiguration clientConfiguration;
    @Resource
    EmqxClient emqxClient;

    @Override
    public void onApplicationEvent(MessageObservable event) {
        if(!event.getSource().getPushChannel().equals(PushChannel.MQTT)){
            return;
        }
        if(!(event.getBody() instanceof Map)){
            return;
        }
        Message message = Message.builder()
                .command(CommandEnum.message)
                .client_msg_id(event.getId())
                .success(true)
                .message("success")
                .send_time(System.currentTimeMillis())
                .req(JSON.parseObject(JSON.toJSONString(event.getBody())))
                .build();
        String body = JSONObject.toJSONString(message);
        String mqttClient = clientConfiguration.getConfig().get(event.getClientId()).getMqtt();
        String topics = "";
        switch (event.getSource().getExtMqttType()){
            case USER:
                topics = Arrays.stream(event.getSource().getTo().split(","))
                        .map(u->MqttMessageType.USER.getUserTopic(active,u)).collect(Collectors.joining(","));
                emqxClient.sendMessageTopics(mqttClient, topics,body);
                return;
            case ALL_USER:
                emqxClient.sendMessageTopic(mqttClient, MqttMessageType.ALL_USER.getTopic(active,event.getClientId()),body);
                return;
            case GROUP:
                topics = Arrays.stream(event.getSource().getTo().split(","))
                        .map(u->MqttMessageType.GROUP.getUserTopic(active,u)).collect(Collectors.joining(","));
                emqxClient.sendMessageTopics(mqttClient, topics,body);
                return;
        }
    }
}
